package com.corn.flink.lesson2;

import cn.hutool.core.io.FileUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2023/2/22 10:57
 */

public class FlinkSourceOperatorDemo {

    public static void main(String[] args) throws Exception {
        LocalStreamEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        // 1. 从文件中读取信息
        String filePath = FileUtil.getAbsolutePath("classpath:input/wordcount.txt");
        DataStreamSource<String> fileDs = env.readTextFile(filePath);
        fileDs.print("file source job");

        // 2. 从socket读取数据源 配合linux nc命令
//        DataStreamSource<String> socketDs = env.socketTextStream("node1", 8888);

        // 3. 从集合中读取数据 开发使用
        List<Integer> intList = IntStream.range(0, 10).boxed().collect(Collectors.toList());
        DataStreamSource<Integer> collectionDs = env.fromCollection(intList);
        collectionDs.print("print source job");

        // 4. 从元素中获取
        DataStreamSource<Integer> elementsDs = env.fromElements(1, 2, 3, 4, 5, 6);
        elementsDs.print("element source job");

        env.execute("sourceDemo");
    }
}
